package tasks.io.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tasks.io.client.ClientChannelContext;
import tasks.io.client.ReconnConf;
import tasks.io.client.TioClientConfig;
import tasks.io.core.Packet.Meta;
import tasks.io.core.lock.SetWithLock;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;


public class Io {


    private static final Logger log = LoggerFactory.getLogger(Io.class);


    /**
     * 阻塞发送消息到指定ChannelContext
     */
    public static Boolean bSend(ChannelContext channelContext, Packet packet) {
        if (channelContext == null) {
            return false;
        }
        CountDownLatch countDownLatch = new CountDownLatch(1);
        return send(channelContext, packet, countDownLatch);
    }

    /**
     * 发送到指定的ip和port
     */
    public static Boolean bSend(TioConfig tioConfig, String ip, int port, Packet packet) {
        return send(tioConfig, ip, port, packet, true);
    }


    /**
     * 发消息给指定ChannelContext id
     */
    public static Boolean bSendToId(TioConfig tioConfig, String channelContextId, Packet packet) {
        return sendToId(tioConfig, channelContextId, packet, true);
    }


    /**
     * 关闭连接
     */
    public static void close(ChannelContext channelContext, String remark) {
        close(channelContext, null, remark);
    }


    public static void close(ChannelContext channelContext, String remark, CloseCode closeCode) {
        close(channelContext, null, remark, closeCode);
    }

    /**
     * 关闭连接
     */
    public static void close(ChannelContext channelContext, Throwable throwable, String remark) {
        close(channelContext, throwable, remark, false);
    }

    public static void close(ChannelContext channelContext, Throwable throwable, String remark, CloseCode closeCode) {
        close(channelContext, throwable, remark, false, closeCode);
    }

    public static void close(ChannelContext channelContext, Throwable throwable, String remark, boolean isNeedRemove) {
        close(channelContext, throwable, remark, isNeedRemove, true);
    }

    public static void close(ChannelContext channelContext, Throwable throwable, String remark, boolean isNeedRemove, CloseCode closeCode) {
        close(channelContext, throwable, remark, isNeedRemove, true, closeCode);
    }

    public static void close(ChannelContext channelContext, Throwable throwable, String remark, boolean isNeedRemove, boolean needCloseLock) {
        close(channelContext, throwable, remark, isNeedRemove, needCloseLock, null);
    }


    public static void close(ChannelContext channelContext, Throwable throwable, String remark, boolean isNeedRemove, boolean needCloseLock, CloseCode closeCode) {
        if (channelContext == null) {
            return;
        }
        if (channelContext.isWaitingClose) {
            log.debug("{} 正在等待被关闭", channelContext);
            return;
        }

        //先立即取消各项任务，这样可防止有新的任务被提交进来
        channelContext.decodeRunnable.setCanceled(true);
        channelContext.handlerRunnable.setCanceled(true);
        channelContext.sendRunnable.setCanceled(true);

        WriteLock writeLock = null;
        if (needCloseLock) {
            writeLock = channelContext.closeLock.writeLock();

            boolean tryLock = writeLock.tryLock();
            if (!tryLock) {
                return;
            }
            channelContext.isWaitingClose = true;
            writeLock.unlock();
        } else {
            channelContext.isWaitingClose = true;
        }

        if (closeCode == null) {
            if (channelContext.getCloseCode() == CloseCode.INIT_STATUS) {
                channelContext.setCloseCode(CloseCode.NO_CODE);
            }
        } else {
            channelContext.setCloseCode(closeCode);
        }

        if (channelContext.asynchronousSocketChannel != null) {
            try {
                channelContext.asynchronousSocketChannel.shutdownInput();
            } catch (Throwable e) {
                //log.error(e.toString(), e);
            }
            try {
                channelContext.asynchronousSocketChannel.shutdownOutput();
            } catch (Throwable e) {
                //log.error(e.toString(), e);
            }
            try {
                channelContext.asynchronousSocketChannel.close();
            } catch (Throwable e) {
                //log.error(e.toString(), e);
            }
        }

        channelContext.closeMetaData.setRemark(remark);
        channelContext.closeMetaData.setThrowable(throwable);
        if (!isNeedRemove) {
            if (channelContext.isServer()) {
                isNeedRemove = true;
            } else {
                ClientChannelContext clientChannelContext = (ClientChannelContext) channelContext;
                if (!ReconnConf.isNeedReconn(clientChannelContext, false)) { //不需要重连
                    isNeedRemove = true;
                }
            }
        }
        channelContext.closeMetaData.setNeedRemove(isNeedRemove);

        channelContext.tioConfig.closeRunnable.add(channelContext);
        channelContext.tioConfig.closeRunnable.execute();
    }

    /**
     * 关闭连接
     */
    public static void close(TioConfig tioConfig, String clientIp, Integer clientPort, Throwable throwable, String remark) {
        ChannelContext channelContext = tioConfig.clientNodes.find(clientIp, clientPort);
        close(channelContext, throwable, remark);
    }


    /**
     * 获取所有连接，包括当前处于断开状态的
     */
    public static SetWithLock<ChannelContext> getAll(TioConfig tioConfig) {
        return tioConfig.connections;
    }

    /**
     * 获取所有连接，包括当前处于断开状态的
     *
     * @deprecated 用getAll(TioConfig tioConfig)
     */
    public static SetWithLock<ChannelContext> getAllChannelContexts(TioConfig tioConfig) {
        return getAll(tioConfig);
    }

    /**
     * 此API仅供 tio client使用
     * 获取所有处于正常连接状态的连接
     */
    public static SetWithLock<ChannelContext> getConnecteds(TioClientConfig tioClientConfig) {
        return tioClientConfig.connecteds;
    }

    /**
     * 此API仅供 tio client使用
     * 获取所有处于正常连接状态的连接
     *
     * @deprecated 用getAllConnecteds(TioClientConfig tioClientConfig)
     */
    public static SetWithLock<ChannelContext> getAllConnectedsChannelContexts(TioClientConfig tioClientConfig) {
        return getConnecteds(tioClientConfig);
    }


    /**
     * 根据clientip和clientport获取ChannelContext
     */
    public static ChannelContext getByClientNode(TioConfig tioConfig, String clientIp, Integer clientPort) {
        return tioConfig.clientNodes.find(clientIp, clientPort);
    }

    /**
     * 根据clientip和clientport获取ChannelContext
     *
     * @deprecated getByClientNode(tioConfig, clientIp, clientPort)
     */
    public static ChannelContext getChannelContextByClientNode(TioConfig tioConfig, String clientIp, Integer clientPort) {
        return getByClientNode(tioConfig, clientIp, clientPort);
    }

    /**
     * 根据ChannelContext.id获取ChannelContext
     *
     * @param channelContextId
     * @return
     * @author tanyaowu
     */
    public static ChannelContext getByChannelContextId(TioConfig tioConfig, String channelContextId) {
        //return tioConfig.ids.find(tioConfig, channelContextId);
        return null;
    }

    /**
     * 根据ChannelContext.id获取ChannelContext
     *
     * @param channelContextId
     * @return
     * @author tanyaowu
     * @deprecated 用getByChannelContextId(tioConfig, channelContextId)
     */
    public static ChannelContext getChannelContextById(TioConfig tioConfig, String channelContextId) {
        return getByChannelContextId(tioConfig, channelContextId);
    }


    public static void remove(ChannelContext channelContext, String remark) {
        remove(channelContext, remark, null);
    }

    /**
     * 和close方法对应，只不过不再进行重连等维护性的操作
     */
    public static void remove(ChannelContext channelContext, String remark, CloseCode closeCode) {
        remove(channelContext, null, remark, closeCode);
    }

    /**
     * 和close方法对应，只不过不再进行重连等维护性的操作
     */
    public static void remove(ChannelContext channelContext, Throwable throwable, String remark) {
        remove(channelContext, throwable, remark, null);
    }

    /**
     * 和close方法对应，只不过不再进行重连等维护性的操作
     */
    public static void remove(ChannelContext channelContext, Throwable throwable, String remark, CloseCode closeCode) {
        close(channelContext, throwable, remark, true, closeCode);
    }

    /**
     * 和close方法对应，只不过不再进行重连等维护性的操作
     */
    public static void remove(TioConfig tioConfig, String clientIp, Integer clientPort, Throwable throwable, String remark) {
        remove(tioConfig, clientIp, clientPort, throwable, remark, null);
    }

    /**
     * 删除clientip和clientPort为指定值的连接
     */
    public static void remove(TioConfig tioConfig, String clientIp, Integer clientPort, Throwable throwable, String remark, CloseCode closeCode) {
        ChannelContext channelContext = tioConfig.clientNodes.find(clientIp, clientPort);
        remove(channelContext, throwable, remark, closeCode);
    }


    /**
     * 发送消息到指定ChannelContext
     */
    public static Boolean send(ChannelContext channelContext, Packet packet) {
        return send(channelContext, packet, null);
    }

    private static Boolean send(final ChannelContext channelContext, Packet packet, CountDownLatch countDownLatch) {
        try {
            if (packet == null || channelContext == null) {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                return false;
            }

            if (channelContext.isVirtual) {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                return true;
            }

            if (channelContext.isClosed || channelContext.isRemoved) {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                if (channelContext != null) {
                    log.info("can't send data, {}, isClosed:{}, isRemoved:{}", channelContext, channelContext.isClosed, channelContext.isRemoved);
                }
                return false;
            }

            if (channelContext.tioConfig.packetConverter != null) {
                Packet packet1 = channelContext.tioConfig.packetConverter.convert(packet, channelContext);
                if (packet1 == null) {
                    if (log.isInfoEnabled()) {
                        log.info("convert后为null，表示不需要发送", channelContext, packet.logstr());
                    }
                    return true;
                }
                packet = packet1;
            }

            boolean isSingleBlock = countDownLatch != null;

            boolean isAdded = false;
            if (countDownLatch != null) {
                Meta meta = new Meta();
                meta.setCountDownLatch(countDownLatch);
                packet.setMeta(meta);
            }

            if (channelContext.tioConfig.useQueueSend) {
                isAdded = channelContext.sendRunnable.add(packet);
            } else {
                isAdded = channelContext.sendRunnable.sendPacket(packet);
            }

            if (!isAdded) {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                return false;
            }
            if (channelContext.tioConfig.useQueueSend) {
                channelContext.sendRunnable.execute();
            }

            if (isSingleBlock) {
                long timeout = 10;
                try {
                    Boolean awaitFlag = countDownLatch.await(timeout, TimeUnit.SECONDS);
                    if (!awaitFlag) {
                        log.error("{}, 阻塞发送超时, timeout:{}s, packet:{}", channelContext, timeout, packet.logstr());
                    }
                } catch (InterruptedException e) {
                    log.error(e.toString(), e);
                }

                Boolean isSentSuccess = packet.getMeta().getIsSentSuccess();
                return isSentSuccess;
            } else {
                return true;
            }
        } catch (Throwable e) {
            log.error(channelContext + ", " + e, e);
            return false;
        }

    }

    /**
     * 发送到指定的ip和port
     */
    public static Boolean send(TioConfig tioConfig, String ip, int port, Packet packet) {
        return send(tioConfig, ip, port, packet, false);
    }

    /**
     * 发送到指定的ip和port
     */
    private static Boolean send(TioConfig tioConfig, String ip, int port, Packet packet, boolean isBlock) {
        ChannelContext channelContext = tioConfig.clientNodes.find(ip, port);
        if (channelContext != null) {
            if (isBlock) {
                return bSend(channelContext, packet);
            } else {
                return send(channelContext, packet);
            }
        } else {
            log.info("{}, can find channelContext by {}:{}", tioConfig.getName(), ip, port);
            return false;
        }
    }


    /**
     * 发消息给指定ChannelContext id
     */
    public static Boolean sendToId(TioConfig tioConfig, String channelContextId, Packet packet) {
        return sendToId(tioConfig, channelContextId, packet, false);
    }

    /**
     * 发消息给指定ChannelContext id
     */
    private static Boolean sendToId(TioConfig tioConfig, String channelContextId, Packet packet, boolean isBlock) {
        ChannelContext channelContext = Io.getChannelContextById(tioConfig, channelContextId);
        if (channelContext == null) {
            return false;
        }
        if (isBlock) {
            return bSend(channelContext, packet);
        } else {
            return send(channelContext, packet);
        }
    }


    private Io() {
    }

}
